package com.atguigu.utils;

import com.atguigu.bean.TransientSink;
import com.atguigu.common.GmallConfig;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class MyClickHouseUtil {

    // clickhouseSink
    public static <T>SinkFunction<T> getClickHouseSink(String sql) {

        return JdbcSink.<T>sink(sql,
                new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {

                        // 通过反射获取对象属性
                        Class<?> cls = t.getClass();
                        // 通过反射拿到所有字段
                        Field[] fields = cls.getDeclaredFields();

                        // 设置第二个指针
                        int offset = 0;
                        // 遍历所有字段
                        for (int i = 0; i < fields.length; i++) {

                            // 将字段的封装权限设置为可获取
                            Field field = fields[i];
                            field.setAccessible(true);

                            // 获取字段上不写出到clickhouse的注解
                            TransientSink annotation = field.getAnnotation(TransientSink.class);
                            if (annotation != null){
                                offset++;
                                continue;
                            }

                            // 通过属性名使用反射拿到对应值
                            Object value = null;
                            try {
                                value = field.get(t);
                            } catch (IllegalAccessException e) {
                                System.out.println();
                                System.out.println();
                                System.out.println();
                                System.out.println();
                                System.out.println();
                                System.out.println();
                                System.out.println("通过反射获取属性值失败！！！！！");
                                System.out.println();
                                System.out.println();
                                System.out.println();
                                System.out.println();
                                System.out.println();
                                e.printStackTrace();
                            }
                            preparedStatement.setObject(i + 1 - offset, value);
                        }
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchIntervalMs(1000L)
                        .withBatchSize(5)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .build());

    }


}
